package com.enjoyor.soa.traffic.frame.support.amq;


import java.util.concurrent.TimeUnit;

import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

import com.enjoyor.soa.traffic.util.helper.LoggerHelper;

/**
 * AMQ信息统计服务，用于AMQ消息队列属性信息统计，具体包括以下几种属性：
 *
 * <p><table><tr align="left">
 * <th bgcolor="#CCCCCC" align="center" id="construct">消息队列属性</th>
 * <th bgcolor="#CCCCCC" align="center" id="matches">说明</th></tr>
 * <tr><td valign="top" headers="construct characters"><i>size</i></td>
 *     <td headers="matches">队列中未消费的数据数量</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>memoryUsage</i></td>
 *     <td headers="matches">可用数据存储空间大小</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>dequeueCount</i></td>
 *     <td headers="matches">已消费的数据数量</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>averageEnqueueTime</i></td>
 *     <td headers="matches">数据平均入列时间</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>destinationName</i></td>
 *     <td headers="matches">当前队列名称</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>averageMessageSize</i></td>
 *     <td headers="matches">平均数据大小</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>memoryPercentUsage</i></td>
 *     <td headers="matches">占用的内存大小（百分比）</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>producerCount</i></td>
 *     <td headers="matches">生产者数量</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>consumerCount</i></td>
 *     <td headers="matches">消费者数量</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>minEnqueueTime</i></td>
 *     <td headers="matches">数据最小入列时间</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>expiredCount</i></td>
 *     <td headers="matches">超时数量</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>dispatchCount</i></td>
 *     <td headers="matches">数据分发数量</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>maxEnqueueTime</i></td>
 *     <td headers="matches">数据最大入列时间</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>enqueueCount</i></td>
 *     <td headers="matches">已入列数据数量</td></tr>
 * <tr><td valign="top" headers="construct characters"><i>memoryLimit</i></td>
 *     <td headers="matches">队列可以使用的最大存储容量</td></tr>
 * </table></p>
 *
 * @author WY
 * @reviser WY
 * @date 2015-4-8
 */

public class AMQStatistics extends AbstractActiveMQ {
	private MessageConsumer consumer;
	private MessageProducer producer;
	private MapMessage      reply;

	/**
	 * 构造消息统计工具
	 *
	 * @param url
	 *            需要统计的消息队列URL
	 * @param queue
	 *            需要统计的消息队列名称
	 */
	public AMQStatistics(String url, String queue) {
		super(url, queue);
	}

	@Override
	public void initialize() {
		if (this.running()) {
			try {
				ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
						getUrl());
				setConnection(connectionFactory.createConnection());
				getConnection().start();
				setSession(getConnection().createSession(Boolean.FALSE,
						Session.AUTO_ACKNOWLEDGE));
			} catch (JMSException e) {
				LoggerHelper.AMQ_LOG.warn(getQueue()
						+ " AMQ消息接收引擎连接异常，正在尝试重连.....");
				try {
					TimeUnit.SECONDS.sleep(CONNECTION_FAILED_DELAY);
				} catch (InterruptedException e1) {
					LoggerHelper.AMQ_LOG.warn(getQueue() + " AMQ消息发送引擎延时异常", e1);
				}
				initialize();
			}
		}
	}

	/**
	 * 获取所选队列的属性信息
	 * @param name 属性名称(具体名称参见本类属性说明)
	 * @return 属性信息，如果队列不存在或属性不存在则返回null
	 */
	public Object getStatistics(String name) {
		Object statistics = null;
		try {
			// 构造用于信息统计的临时队列
			Queue replyTo = getSession().createTemporaryQueue();
			consumer = getSession().createConsumer(replyTo);
			Queue testQueue = getSession().createQueue(getQueue());
			producer = getSession().createProducer(null);
			String queueName = "ActiveMQ.Statistics.Destination."
					+ testQueue.getQueueName();
			Queue query = getSession().createQueue(queueName);
			Message msg = getSession().createMessage();
			producer.send(testQueue, msg);
			msg.setJMSReplyTo(replyTo);
			producer.send(query, msg);
			// 获取统计属性信息
			reply = (MapMessage) consumer.receive();

			if ((reply == null) || (!reply.getMapNames().hasMoreElements())) {
				LoggerHelper.AMQ_LOG.warn("This queue(" + getQueue()
						+ ") has no statistics message.");
				return null;
			}
			statistics = reply.getObject(name);
		} catch (JMSException e) {
			LoggerHelper.AMQ_LOG.warn(getQueue() + " AMQ消息接收引擎连接异常，正在尝试重连.....");
			endStatistics();
			try {
				TimeUnit.SECONDS.sleep(CONNECTION_FAILED_DELAY);
			} catch (InterruptedException e1) {
				LoggerHelper.AMQ_LOG.warn(getQueue() + " AMQ消息发送引擎延时异常", e1);
			}
			initialize();
		}

		return statistics;
	}

	@Override
	public void endEngine() {
		endStatistics();
		this.shutdown();
	}

	private void endStatistics() {
		try {
			this.consumer.close();
			this.consumer = null;
			this.producer.close();
			this.producer = null;
			this.getSession().close();
			this.setSession(null);
			this.getConnection().close();
			this.setConnection(null);
		} catch (JMSException e) {
			LoggerHelper.LOG.error("ActiveMQ.Statistics.Destination."
					+ getQueue() + " AMQ消息接收引擎关闭异常", e);
		}
	}

	@Override
	public Message receiveMessage(int time) {
		// TODO Auto-generated method stub
		return null;
	}
}
